package com.ehotting.edsta.mqbd.config;

import com.ehotting.edsta.mqbd.api.enums.SysModuleEnum;
import com.ehotting.edsta.mqbd.listener.ApplicationContextListener;
import com.ehotting.mqbd.api.MQDBConstants;
import com.ehotting.mqbd.api.bean.MqApplicationContext;
import com.ehotting.mqbd.api.bean.solver.ModuleInfoBean;
import com.ehotting.mqbd.api.enums.DirectionEnum;
import com.ehotting.mqbd.api.enums.SolveStepEnum;
import com.ehotting.mqbd.api.enums.TopicEnum;
import com.ehotting.mqbd.core.service.MQService;
import com.ehotting.mqbd.core.config.MqbdCorePropertyConfig;
import com.ehotting.mqbd.core.helper.IPHelper;
import com.ehotting.mqbd.core.helper.MQStringHelper;
import com.ehotting.mqbd.core.logger.MQSendLogger;
import com.ehotting.mqbd.core.logger.MqMsgLogger;
import com.ehotting.mqbd.core.mq.MessageProducer;
import com.ehotting.mqbd.core.mq.batch.BatchConsumerRequestAction;
import com.ehotting.mqbd.core.mq.batch.BatchMessageRequestConsumer;
import com.ehotting.mqbd.core.mq.batch.BatchMqApplicationContext;
import com.ehotting.mqbd.core.mq.consumer.MessageConsumer;
import com.ehotting.mqbd.core.manager.ModuleManager;
import com.ehotting.mqbd.core.manager.SolverAllocator;
import com.ehotting.mqbd.core.manager.SolverInfoManager;
import com.ehotting.mqbd.core.sync.MqSyncFutureMapHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

@Configuration
@DependsOn("mqbdCorePropertyConfig")
public class MqbdConsumerCommonCfg {

    private static Logger logger = LoggerFactory.getLogger(MqbdConsumerCommonCfg.class);

    @Autowired
    private IPHelper ipHelper;

    @Bean(name="batchMqApplicationContext")
    public BatchMqApplicationContext batchMqApplicationContext(ModuleManager moduleManager){
        List<MqApplicationContext> mqApplicationContextList = Arrays.stream(SysModuleEnum.values()).map(module->{
            MqApplicationContext mqApplicationContext = new MqApplicationContext();
            mqApplicationContext.setModule(module.getCode());
            ModuleInfoBean protocol = moduleManager.getByModule(mqApplicationContext.getModule());
            logger.info("module:{},protocol:{}",mqApplicationContext.getModule(),protocol);
            if(protocol!=null && protocol.getSolveStep() == SolveStepEnum.MULTI){
                mqApplicationContext.setLocalHost(ipHelper.getIp());
            }else{
                mqApplicationContext.setLocalHost(MQDBConstants.DISPATCHER_HOST);
            }
            mqApplicationContext.setConsumerHost(MQDBConstants.DISPATCHER_HOST);
            return mqApplicationContext;
        }).collect(Collectors.toList());

        BatchMqApplicationContext batchMqApplicationContext = new BatchMqApplicationContext();
        batchMqApplicationContext.getMqApplicationContextList().addAll(mqApplicationContextList);
        return batchMqApplicationContext;
    }


    @DependsOn("mqbdCorePropertyConfig")
    @Bean(name ="messageProducer",initMethod = "start",destroyMethod = "shutdown")
    public MessageProducer concurrencyMessageProducer(MqbdCorePropertyConfig mqbdCorePropertyConfig)  {
        MessageProducer messageProducer = new MessageProducer(
                mqbdCorePropertyConfig.getProducer(),
                MQStringHelper.createProducerGroup(MQDBConstants.DISPATCHER_HOST));
        return messageProducer;
    }


    @Bean(name="mqService")
    public MQService mqService(MessageProducer messageProducer,MqSyncFutureMapHelper mqSyncFutureMapHelper,
                               MQSendLogger mqSendLogger){
        return new MQService(messageProducer,mqSyncFutureMapHelper,mqSendLogger);
    }

    @DependsOn("mqbdCorePropertyConfig")
    @Bean(initMethod = "start",destroyMethod = "shutdown")
    public BatchMessageRequestConsumer concurrencyRequestMessageConsumer(MqbdCorePropertyConfig mqbdCorePropertyConfig,
                                                                         BatchConsumerRequestAction concurrencyConsumerRequestBatchAction){

        BatchMessageRequestConsumer messageConsumer = new BatchMessageRequestConsumer(
                mqbdCorePropertyConfig.getConsumer(),
                concurrencyConsumerRequestBatchAction);

        return messageConsumer;
    }

    @Bean(initMethod = "start",destroyMethod = "shutdown")
    public BatchMessageRequestConsumer blockRequestMessageConsumer(MqbdCorePropertyConfig mqbdCorePropertyConfig,
                                                                   BatchConsumerRequestAction blockConsumerRequestBatchAction){

        MessageConsumer.MessageConsumerParam consumerParam = mqbdCorePropertyConfig.getConsumer().clone();
        consumerParam.setConsumeThreadMin(1);
        consumerParam.setConsumeMessageBatchMaxSize(1);

        BatchMessageRequestConsumer messageConsumer = new BatchMessageRequestConsumer(
                mqbdCorePropertyConfig.getConsumer(),
                blockConsumerRequestBatchAction);

        return messageConsumer;
    }



    @Bean(name="blockConsumerRequestBatchAction")
    public BatchConsumerRequestAction blockConsumerRequestBatchAction(
            SolverAllocator solverAllocator,
            BatchMqApplicationContext  batchMqApplicationContext,
            SolverInfoManager solverInfoManager,ModuleManager moduleManager, MqMsgLogger mqMsgLogger){
        BatchConsumerRequestAction action = new BatchConsumerRequestAction(solverAllocator,
                batchMqApplicationContext,
                TopicEnum.BLOCK,
                DirectionEnum.REQUEST,solverInfoManager,moduleManager,mqMsgLogger);
        return action;
    }


    @Bean(name="concurrencyConsumerRequestBatchAction")
    public BatchConsumerRequestAction concurrencyConsumerRequestBatchAction(SolverAllocator solverAllocator,
                                                                            BatchMqApplicationContext batchMqApplicationContext
            , SolverInfoManager solverInfoManager,
                                                                            ModuleManager moduleManager, MqMsgLogger mqMsgLogger){
        BatchConsumerRequestAction action = new BatchConsumerRequestAction(solverAllocator,
                batchMqApplicationContext,
                TopicEnum.CONCURRENCY,
                DirectionEnum.REQUEST,solverInfoManager,moduleManager,mqMsgLogger);
        return action;
    }

}